package cn.ctodb.im.kafka;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.internals.RecordHeader;

import java.io.IOException;

/**
 * All rights Reserved, Designed By www.ctodb.cn
 *
 * @version V1.0
 * @author: lichaohn@163.com
 * @date: 2018-11-16 13:37
 * @Copyright: 2018 www.ctodb.cn All rights reserved.
 */
public class KafkaUtils {

    private static ObjectMapper objectMapper = new ObjectMapper();
    private static final String channelId = "channelId";
    private static final String dataClass = "dataClass";

    public static ProducerRecord toProducerRecord(String topic, String channelId, Object data) throws JsonProcessingException {
        String str = objectMapper.writeValueAsString(data);
        ProducerRecord producerRecord = new ProducerRecord<String, String>(topic, str);
        producerRecord.headers().add(new RecordHeader(dataClass, data.getClass().getName().getBytes())).add(new RecordHeader(channelId, channelId.getBytes()));
        return producerRecord;
    }

    public static String getChannelId(ConsumerRecord<?, String> record) {
        for (Header header : record.headers()) {
            if (channelId.equals(header.key())) {
                return new String(header.value());
            }
        }
        return null;
    }

    public static <T> T getDate(ConsumerRecord<?, String> record) throws ClassNotFoundException, IOException {
        for (Header header : record.headers()) {
            if (dataClass.equals(header.key())) {
                Class c = Class.forName(new String(header.value()));
                return (T) objectMapper.readValue(record.value(), c);
            }
        }
        return null;
    }

}
